#include "xNetThread.h"
//extern void AddSession(xNet::xTCPTask* task);
namespace xNet
{
    xNetThread::xNetThread()
               : m_epFd(-1)
               , m_eventSize(MAX_EPOLL_EVENTS)
    {
        m_epFd = epoll_create(128);
        m_events.resize(MAX_EPOLL_EVENTS);
    }
    xNetThread::~xNetThread()
    {
        for(auto it=m_que.begin();it!=m_que.end();++it)
        {
            xTCPTask* task = *it;
            XLOG_INFO("TCP Session Not Handle,Close fd:[%d]",task->GetTCPSock()->GetFd());
            DELETE(task);
        }
        m_que.clear();
    }

    bool xNetThread::Run()
    {
        while(!GetStart());
        while(!GetTermnation())
        {
            this->CheckState();
            int activeSize = epoll_wait(m_epFd,&m_events[0],m_eventSize,MAX_EPOLL_THREAD_TIMEOUT);
            if(GetTermnation())
            {
                break;
            }
            for(int i=0;i<activeSize;++i)
            {
                xTCPTask* task = static_cast<xTCPTask*>(m_events[i].data.ptr);
                if(m_events[i].events & EPOLLIN)
                {
                    task->ListenRecv();
                }
                else if(m_events[i].events & EPOLLOUT)
                {
                    task->ListenSend();
                }
            }
        }
        return true;
    }

    void xNetThread::Finish()
    {
        xThread::Finish();
        if(m_epFd != -1)
        {
            close(m_epFd);
            m_epFd = -1;
        }
        for(auto it=m_tasksDeque.begin();it!=m_tasksDeque.end();++it)
        {
            xTCPTask* task = *it;
            XLOG_INFO("TCP Session Handled,Close fd:[%d]",task->GetTCPSock()->GetFd());
            DELETE(task);
        }
        m_tasksDeque.clear();
    }

    void xNetThread::CheckState()
    {
        {
            std::lock_guard<std::mutex> guard(m_mutex);
            for(auto it=m_que.begin();it!=m_que.end();++it)
            {
                m_tasksDeque.push_back(*it);
            }
            m_que.clear();
        }

        for(auto it=m_tasksDeque.begin();it!=m_tasksDeque.end();)
        {
            xTCPTask* task = *it;
            if(task->GetState() == TCP_TASK_STATE_NORMAL)
            {
                task->GetTCPSock()->AddEPOLL(m_epFd,task);
                task->SetState(TCP_TASK_STATE_VERIFY);
                task->AddSession(task);
                ++it;
            }
            else if(task->GetState() == TCP_TASK_STATE_RECYCLE 
                    || task->GetState() == TCP_TASK_STATE_EXITING)
            {
                task->GetTCPSock()->DelEPOLL(m_epFd);
                it = m_tasksDeque.erase(it);
                task->SetState(TCP_TASK_STATE_EXIT);
            }
            else 
            {
                ++it;
            }
        }
        if(m_tasksDeque.size() > m_eventSize)
        {
            m_eventSize = m_tasksDeque.size();
            m_events.resize(m_eventSize);
        }
    }
}